package tjs.study.notes.flink.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.commons.math3.random.RandomDataGenerator;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerTest {
    public static final String broker_list = "localhost:9092";
    public static final String topic = "tjs_test_cousumer";  // kafka topic，Flink 程序中需要和这个统一
    private static Integer id = 0;
    private static RandomDataGenerator randomDataGenerator = new RandomDataGenerator();

    public static void writeToKafka() throws InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", broker_list);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //key 序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //value 序列化
        org.apache.kafka.clients.producer.KafkaProducer producer = new org.apache.kafka.clients.producer.KafkaProducer<String, String>(props);

        Integer curId = id++;
        Student student = new Student("nicky" + curId, "password" + curId, randomDataGenerator.nextInt(0, 5), curId);
        ProducerRecord record = new ProducerRecord<String, String>(topic, null, null, JSON.toJSONString(student));
        producer.send(record);
        System.out.println("发送数据: " + JSON.toJSONString(student));

        producer.flush();
    }

    public static void main(String[] args) throws InterruptedException {
        int i = 0;
        while (i++ < 1000) {
            Thread.sleep(100);
            writeToKafka();
        }
    }
}
